package cn.boat.router.producer;

import cn.boat.router.MetaZookeeper;
import cn.boat.router.broker.BrokerZookeeper;
import cn.boat.router.broker.partition.strategy.PartitionSelector;
import cn.boat.router.broker.partition.strategy.impl.PollingPartitionSelector;
import cn.boat.router.exception.InvalidArguementBoatMQException;
import cn.boat.router.models.Broker;
import cn.boat.router.models.ConfigCenter;
import cn.boat.router.models.StateModel;
import cn.boat.router.utils.ConvertUtils;
import cn.boat.router.utils.JSONUtils;
import cn.boat.router.utils.PropUtils;
import cn.boat.router.utils.ZkClientUtils;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class ProducerZookeeper implements ZkClientChangedListener {

    private Logger logger = LoggerFactory.getLogger(ProducerZookeeper.class);

    private static Random random = new Random();

    private ZkClient zkClient;

    private MetaZookeeper metaZookeeper;

    public ProducerZookeeper(ZkClient zkClient, MetaZookeeper metaZookeeper) {
        this.zkClient = zkClient;
        this.metaZookeeper = metaZookeeper;
    }

    public void registerTopic(String topicName, int numOfPartition, int numOfBackup) throws Exception {
        registerTopic(topicName, numOfPartition, numOfBackup, new PollingPartitionSelector());
    }

    /**
     * 发布主题
     * @param topicName 主题名
     * @param numOfPartition 分区数
     * @param numOfBackup 分区备份数
     * @param selector 选择策略
     */
    public void registerTopic(String topicName, int numOfPartition, int numOfBackup, PartitionSelector selector) throws Exception {
        // 获取broker数量
        int brokerNum = ZkClientUtils.getChildren(zkClient, metaZookeeper.brokerIdsPath).size();
        if (numOfPartition > brokerNum) {
            throw new InvalidArguementBoatMQException("分区备份数不能超过Broker数量！");
        }

        // 分区基路径
        String newTopicPath = metaZookeeper.brokerTopicsPath+"/"+topicName+"/partitions";

        // 遍历分区设置
        scanPartitions(numOfPartition, newTopicPath, selector, numOfBackup);

        // 监听broker变化
        zkClient.subscribeChildChanges(metaZookeeper.brokerIdsPath, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {
                zkClient.deleteRecursive(metaZookeeper.brokerTopicsPath+"/"+topicName);

                // 遍历分区设置
                scanPartitions(numOfPartition, newTopicPath, selector, numOfBackup);
            }
        });
    }

    private void scanPartitions(int numOfPartition, String newTopicPath, PartitionSelector selector, int numOfBackup) throws Exception {
        for (int i=0; i < numOfPartition; i++) {
            ZkClientUtils.makeSurePersistentPathExists(zkClient, newTopicPath+"/"+i+"/state");

            // 获取所有的broker
            List<Broker> allBrokerData = new BrokerZookeeper(zkClient, new MetaZookeeper(zkClient, "/boatMQ")).getAllBrokerData();
            List<Integer> allBrokerIdData = ConvertUtils.filterBrokerId(allBrokerData);

            // 从broker中选取备份分布
            List<Integer> backupDistribution = selector.getPartitionDistribution(allBrokerIdData, numOfBackup);
            // 从备份分布中选举一份作为leader
            List<Integer> tempList = new ArrayList<>(backupDistribution);
            int leader = selector.getPartitionDistribution(tempList, 1).get(0);

            StateModel stateModel = new StateModel(leader, backupDistribution);
            String stateData = JSONUtils.serializeObject(stateModel);
            ZkClientUtils.updatePersistentPath(zkClient, newTopicPath+"/"+i+"/state", stateData);
        }
    }

    public void setConfigCenter() throws Exception {
        ConfigCenter configCenter = new ConfigCenter();
        configCenter.setDriver("com.mysql.jdbc.Driver");
        configCenter.setUrl("jdbc:mysql://localhost:3306/order?useUnicode=true&amp;characterEncoding=UTF-8");
        configCenter.setUsername("root");
        configCenter.setPassword("root");
        ZkClientUtils.createPersistentPath(zkClient, metaZookeeper.configCenterPath, JSONUtils.serializeObject(configCenter));
    }

    @Override
    public void onZkClientChanged(ZkClientUtils utils) {

    }

    /**
     * 最好只用一次，返回的对象放在主类的一个对象中
     * 暂时从zookeeper.properties里加载
     *
     * @return
     */
    // 单元测试要是通不过,请修改resources/zookeeper.properties里的zk地址
    private ZkClientUtils.ZKConfig loadZkConfigFromLocalFile() {
        try {
            final Properties properties = PropUtils.getResourceAsProperties("zookeeper.properties", "GBK");
            final ZkClientUtils.ZKConfig zkConfig = new ZkClientUtils.ZKConfig();
            if (StringUtils.isNotBlank(properties.getProperty("zk.zkConnect"))) {
                zkConfig.zkConnect = properties.getProperty("zk.zkConnect");
            }

            if (StringUtils.isNotBlank(properties.getProperty("zk.zkSessionTimeoutMs"))) {
                zkConfig.zkSessionTimeoutMs = Integer.parseInt(properties.getProperty("zk.zkSessionTimeoutMs"));
            }

            if (StringUtils.isNotBlank(properties.getProperty("zk.zkConnectionTimeoutMs"))) {
                zkConfig.zkConnectionTimeoutMs = Integer.parseInt(properties.getProperty("zk.zkConnectionTimeoutMs"));
            }

            if (StringUtils.isNotBlank(properties.getProperty("zk.zkSyncTimeMs"))) {
                zkConfig.zkSyncTimeMs = Integer.parseInt(properties.getProperty("zk.zkSyncTimeMs"));
            }

            return zkConfig;
        } catch (final IOException e) {
            logger.error("zookeeper配置失败", e);
            return null;
        }
    }

    public void close(){
        zkClient.close();
        logger.info("zookeeper 连接安全关闭");
    }

    public static void main(String[] args) throws Exception {
        // 该对象放到客户端配置类中
        ZkClientUtils.ZKConfig zkConfig = new ZkClientUtils.ZKConfig();

        ZkClient zkClient =
                new ZkClient(zkConfig.zkConnect, zkConfig.zkSessionTimeoutMs,
                        zkConfig.zkConnectionTimeoutMs, new ZkClientUtils.StringSerializer());

        ProducerZookeeper pz = new ProducerZookeeper(zkClient, new MetaZookeeper(zkClient, "/boatMQ"));

        System.out.println("测试");


        //pz.registerTopic("scanPartitions", 3, 3);

    }
}
